package io.kiki.sba.registry.server.metadata.lease.impl;

import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.metaserver.Lease;
import io.kiki.sba.registry.common.model.metaserver.cluster.VersionedList;
import io.kiki.sba.registry.observer.impl.AbstractLifecycleObservable;
import io.kiki.sba.registry.util.DatumVersionUtil;
import lombok.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;


public class LeaseManager<T extends Node> extends AbstractLifecycleObservable implements io.kiki.sba.registry.server.metadata.lease.LeaseManager<T> {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    protected final AtomicLong currentEpoch = new AtomicLong();
    // Map[ip-address, Lease{node, duration, timestamp}]
    protected final ConcurrentMap<String, Lease<T>> ipToLeaseMap = new ConcurrentHashMap<>();

    protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    protected void register(@NonNull Lease<T> lease) {
        logger.info("[register] register node: {}", lease);
        lock.writeLock().lock();
        try {
            String nodeIp = lease.getRenewal().getUrl().getIp();
            ipToLeaseMap.putIfAbsent(nodeIp, lease);
            refreshEpoch(DatumVersionUtil.nextId());
        } finally {
            lock.writeLock().unlock();
        }
    }

    protected boolean cancel(Lease<T> lease) {
        if (lease == null) {
            throw new IllegalArgumentException("[cancel]NullPointer of renewal");
        }
        // read lock for concurrent modification, and mutext for renew/register operations
        lock.writeLock().lock();
        boolean result = true;
        try {
            logger.info("[cancel][begin] {}", lease);
            result = ipToLeaseMap.remove(lease.getRenewal().getUrl().getIp(), lease);
            logger.info("[cancel][end] {}", result);
            refreshEpoch(DatumVersionUtil.nextId());
        } finally {
            lock.writeLock().unlock();
        }
        return result;
    }

    @Override
    public boolean renew(@NonNull T renewal, int leaseDuration) {
        int validLeaseDuration = leaseDuration > 0 ? leaseDuration : Lease.DEFAULT_DURATION_SECS;

        lock.writeLock().lock();
        try {
            Lease<T> lease = ipToLeaseMap.get(renewal.getUrl().getIp());
            if (lease == null) {
                logger.warn("[renew][node not exist, register: {}-{}]", renewal.getNodeType(), renewal.getUrl().getIp());
                register(new Lease<>(renewal, validLeaseDuration));
                return false;
            }
            logger.info("[renew][renew lease] node: {}-{}, extends: {}s", renewal.getNodeType(), renewal.getUrl().getIp(), validLeaseDuration);

            lease.renew(validLeaseDuration);
        } finally {
            lock.writeLock().unlock();
        }
        return true;
    }

    @Override
    public Lease<T> getLease(T renewal) {
        return ipToLeaseMap.get(renewal.getUrl().getIp());
    }

    @Override
    public VersionedList<Lease<T>> VersionedLeaseList() {
        lock.readLock().lock();
        try {
            return new VersionedList<>(getLeaseList(), currentEpoch.get());
        } finally {
            lock.readLock().unlock();
        }
    }


    protected boolean refreshEpoch(long newEpoch) {
        if (currentEpoch.get() < newEpoch) {
            logger.info("[refreshEpoch] epoch change from {} to {}", currentEpoch.get(), newEpoch);
            currentEpoch.set(newEpoch);
            return true;
        } else {
            logger.info("[refreshEpoch] epoch change not able, current {}, request {}", currentEpoch.get(), newEpoch);
            return false;
        }
    }

    private List<Lease<T>> getLeaseList() {
        List<Lease<T>> list = new ArrayList<>();
        try {
            lock.readLock().lock();
            list.addAll(ipToLeaseMap.values());
            return list;
        } finally {
            lock.readLock().unlock();
        }
    }
}
